package com.junbao.message.infrastructure.queue;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.junbao.message.infrastructure.queue.util.TaskAttributeUtil;

/**
 * @ClassName WheelQueue
 * @Description 像轮子一样转动的队列； 环形队列；循环队列。 
 * @date 2017-03-20
 * @version 1.0.0
 *
 */
public class WheelQueue {
	
	private static final Logger LOG = LoggerFactory.getLogger(WheelQueue.class);
	
	private static final int DEFULT_QUEUE_SIZE = 3600;
	
	/** 建立一个有3600个槽位的环形队列； 每秒轮询一个槽位，3600个就是3600秒=1小时 */
	final Slot[] slotQueue = new Slot[DEFULT_QUEUE_SIZE];
	
	/** 记录当前已经被处理位置*/
	private AtomicInteger currentSlot;
	
	/** 初始时间 记录队列初始化时间，用于时间进入下一个小时的判断*/
	private AtomicReference<Date> initTime;
	
	/**索引偏移量*/
	private static final int OFFSET_INDEX = 1;
	
	/** 任务Id对应的槽位等任务属性 */
	final Map<String, TaskAttribute> taskSlotMapping = new ConcurrentHashMap<String, TaskAttribute>(1000, 1F);

	{
		for (int i = 0; i < DEFULT_QUEUE_SIZE; i++) {
			this.slotQueue[i] = new Slot();
		}
	}
	
	public WheelQueue(int currentSlot){
		this.currentSlot = new AtomicInteger(currentSlot-OFFSET_INDEX);
		this.initTime = new AtomicReference<Date>(new Date());
	}
	/**
	 * 添加一个任务到环形队列
	 * 
	 * @param task  任务
	 * @param secondsLater  以当前时间点为基准，多少毫秒以后执行
	 */
	public void add(final AbstractTask task, int secondsLater) {
		//设置任务属性
		int soltIndex = TaskAttributeUtil.setAttribute(secondsLater, task, taskSlotMapping);
		//加到对应槽位的集合中
		SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss SSS");
		LOG.info("task id "+ task.getId()+" add time is : "+ dateFormat.format(new Date()));
		slotQueue[soltIndex].add(task);
		LOG.debug("join task. task={}, slotIndex={}", task.toString() + soltIndex);
	}
	
	/**
	 * 添加一个任务到环形队列
	 * 
	 * @param task  任务
	 */
	public void add(final AbstractTask task) {
		//设置任务属性
		int soltIndex = TaskAttributeUtil.setAttribute(task.getSecondsLater(), task, taskSlotMapping);
		//加到对应槽位的集合中
		slotQueue[soltIndex].add(task);
		LOG.debug("join task. task={}, slotIndex={}", task.toString() + soltIndex);
	}

	private final ReentrantLock lock = new ReentrantLock();
	
	/**
	 * 根据指定索引获取槽位中的数据。但不删除。
	 * 由于可能在运行中停顿超过1s，导致错过需要遍历的slot
	 * @param currentSlot 当前时间ms
	 * @return	
	 */
	public Slot[] peek(final int currentSlot) {
		try {
			lock.lock();
			int oldSlot = this.currentSlot.getAndSet(currentSlot);
			/**
			 * 如果当第二个小时过来的时候，当前时间的ms值会小于最后放入的旧的时间ms值。
			 */
			if (currentSlot < oldSlot) {
				/** 判断当前时间是否大于旧的时间*/
				if (compareCurrentTimeAndInitTime()) {
					this.currentSlot.set(0);
					oldSlot = 0;
				}else {
					LOG.error("参数异常！currentSlot必须大于等于oldSlot!");
					throw new IllegalAccessError("参数异常！currentSlot必须大于等于oldSlot!");
				}
			}
			LinkedList<Slot> waitHandleSlots = new LinkedList<>();
			for (int index = oldSlot+OFFSET_INDEX; index <= currentSlot; index++) {
				waitHandleSlots.add(slotQueue[index]);
			}
			Slot[] slots = new Slot[waitHandleSlots.size()];
			for (int i = 0 ; i < waitHandleSlots.size() ; i++) {
				slots[i] = waitHandleSlots.get(i);
			}
			return slots;
		} finally{
			lock.unlock();
		}
		
	}

	/**
	 * 根据任务id移除一个任务
	 * 
	 * @param taskId   任务id
	 */
	public void remove(String taskId) {
		TaskAttribute taskAttribute = taskSlotMapping.get(taskId);
		if (taskAttribute != null) {
			slotQueue[taskAttribute.getSoltIndex()].remove(taskId);
		}
	}

	public Map<String, TaskAttribute> getTaskSlotMapping() {
		return taskSlotMapping;
	}
	
	/**
	 * 判断队列初始时间是否大于当前时间
	 * @return
	 */
	private boolean compareCurrentTimeAndInitTime(){
		boolean compareResult = false;
		Date initTime = this.initTime.get();
		Date currentTime = new Date();
		/** 初始时间大于当前时间*/
		if (initTime.getTime() < currentTime.getTime()) {
			compareResult =true;
			this.initTime.set(currentTime);
		}
		return compareResult;
	}
}
